package com.cnostar.tracker.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * rocket 发送消息封装类
 * @Date 2020/3/31 14:28
 * @Author ZhuFeng
 * @Version 2.0
 */
@Component
@Slf4j
public class RocketMqProducer {

    /**
     * 使用RocketMq的生产者
     */
    @Resource
    private DefaultMQProducer producer;


    @Value("${rocketmq.consumer.topics}")
    private String topic;

    /**
     * 同步发送消息（topic：若为空，默认取config.properties中producer.group.topic的值）
     * @param message
     * @return: org.apache.rocketmq.client.producer.SendResult
     * @author: XuWen
     * @date: 2019/8/24 12:22
     */
    public SendResult syncSend(Message message) {
        SendResult sendResult;
        try {
            if (Objects.isNull(message)) {
                throw new IllegalArgumentException("同步消息发送参数异常, Message对象不能为空");
            }
            if (StringUtils.isBlank(message.getTopic())) {
                message.setTopic(topic);
            }
            sendResult = producer.send(message);
            log.info("消息发送成功, 结果:{}", sendResult);
        } catch (Exception e) {
            log.error("同步消息发送参数异常, ", e);
            return null;
        }
        return sendResult;
    }

    /**
     * 同步发送消息
     * @param topic 主题
     * @param tag   标签
     * @param msg   消息内容
     * @return: org.apache.rocketmq.client.producer.SendResult
     * @author: XuWen
     * @date: 2019/8/24 20:23
     */
    public SendResult syncSend(String topic, String tag, String msg) {
        try {
            Message message = new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            return syncSend(message);
        } catch (Exception e) {
            log.error("同步消息发送参数异常, ", e);
        }
        return null;
    }

    /**
     * 同步发送消息
     * @param tag 标签
     * @param msg 消息内容
     * @return: org.apache.rocketmq.client.producer.SendResult
     * @author: XuWen
     * @date: 2019/8/24 20:23
     */
    public SendResult syncSend(String tag, String msg) {
        return syncSend(topic, tag, msg);
    }

    /**
     * 异步发送消息（topic：若为空，默认取config.properties中producer.group.topic的值）
     * @param message
     * @return: void
     * @author: XuWen
     * @date: 2019/8/24 12:22
     */
    public void asyncSend(Message message) {
        try {
            if (Objects.isNull(message)) {
                throw new IllegalArgumentException("异步消息发送参数异常, Message对象不能为空");
            }
            if (StringUtils.isBlank(message.getTopic())) {
                message.setTopic(topic);
            }
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步消息发送成功, 结果:{}", sendResult);
                }

                @Override
                public void onException(Throwable t) {
                    log.error("异步消息发送参数异常", t);
                }
            });
        } catch (Exception e) {
            log.error("异步消息发送参数异常, ", e);
        }
    }

    /**
     * 异步发送消息
     * @param topic 主题
     * @param tag   标签
     * @param msg   消息内容
     * @return: void
     * @author: XuWen
     * @date: 2019/8/24 20:23
     */
    public void asyncSend(String topic, String tag, String msg) {
        try {
            Message message = new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            asyncSend(message);
        } catch (Exception e) {
            log.error("同步消息发送参数异常, ", e);
        }
    }

    /**
     * 异步发送消息
     * @param tag 标签
     * @param msg 消息内容
     * @return: void
     * @author: XuWen
     * @date: 2019/8/24 20:23
     */
    public void asyncSend(String tag, String msg) {
        asyncSend(topic, tag, msg);
    }


    /**
     * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
     * 此方法在netty启动main方法exitHook中进行触发
     */
    public void close() {
        if (producer != null) {
            log.info("关闭RokectMQ生产者");
            producer.shutdown();
        }
    }
}
